package org.apache.rocketmq.proxy.grpc.v2.channel;

import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.utils.StartAndShutdown;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.config.ProxyConfig;
import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
import org.apache.rocketmq.proxy.service.relay.ProxyRelayResult;
import org.apache.rocketmq.proxy.service.relay.ProxyRelayService;
import org.apache.rocketmq.remoting.protocol.ResponseCode;

import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;


/**
 * @方法描述：grpcchannel管理器，与之相对的是RemotingChannelManager管理器，因为我们构建的proxy模块既可以使用Netty服务端启动
 * 也可以使用grpc服务端启动，客户端既可以访问Netty服务端，也可以访问grpc服务端，所以在proxy模块内部就定义了两个channel管理器
 * 访问netty服务端的客户端channel会被RemotingChannelManager管理器管理，访问grpc服务端的客户端channel会被GrpcChannelManager管理器管理
 * 当然，在我们自己的代码中，我就不为大家引入RemotingChannelManager管理器了，毕竟rocketmq的5.0客户端已经使用grpc和proxy进行通信了，所以我就只实现grpc的channel管理器了
 */
public class GrpcChannelManager implements StartAndShutdown {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
    //这个成员变量就是代理服务，光这样解释大家肯定不明白这个类的对象该发挥什么作用，以及怎么发挥作用
    //请大家不要着急，看到ProxyChannel的时候，大家就清楚这个成员变量的作用了
    private final ProxyRelayService proxyRelayService;
    //客戶端管理器
    private final GrpcClientSettingsManager grpcClientSettingsManager;
    //缓存grpcchannel的表，proxy的grpc服务端每和一个新的客户端建立了连接，就会在服务端内部为这个客户端创建一个GrpcClientChannel对象
    //然后把这个GrpcClientChannel对象放入缓存中，key为clientId客户端唯一Id，value为对应的GrpcClientChannel对象
    protected final ConcurrentMap<String, GrpcClientChannel> clientIdChannelMap = new ConcurrentHashMap<>();

    //为responseFuture对象分配Id的生成器
    //现在大家还不知道responseFuture究竟是什么，还是别急，等到了ProxyChannel中一看就知道了
    protected final AtomicLong nonceIdGenerator = new AtomicLong(0);
    //Grpc通道管理器
    protected GrpcChannelManager grpcChannelManager;
    //缓存所有responseFuture的map，key就是上面的nonceIdGenerator分配的唯一Id
    //我现在能跟大家说的是，生产者和消费者客户端和proxy模块建立连接之后，客户端就可以一直向服务端发送消息了
    //那假如proxy服务端向主动向客户端发送消息呢？这时候该怎么办呢？显然就可以使用服务端为客户端创建的GrpcClientChannel来发送消息
    //而proxy服务端发送了请求之后，等待响应的时候，就会用到responseFuture对象了，这个responseFuture对象会封装服务端接收到的客户端响应
    protected final ConcurrentMap<String /* nonce */, ResultFuture> resultNonceFutureMap = new ConcurrentHashMap<>();
    //定期扫描resultNonceFutureMap响应表的执行器，处理响应表中响应超时的responseFuture对象
    protected final ScheduledExecutorService scheduledExecutorService = ThreadUtils.newSingleThreadScheduledExecutor(
            new ThreadFactoryImpl("GrpcChannelManager_")
    );
    //构造方法
    public GrpcChannelManager(ProxyRelayService proxyRelayService, GrpcClientSettingsManager grpcClientSettingsManager) {
        this.proxyRelayService = proxyRelayService;
        this.grpcClientSettingsManager = grpcClientSettingsManager;
        // 调用init方法进行初始化
        this.init();
    }
    //初始化方法
    protected void init() {
        //在该方法中向执行器提交了定时扫描响应表的任务
        this.scheduledExecutorService.scheduleAtFixedRate(
                this::scanExpireResultFuture,
                10, 1, TimeUnit.SECONDS
        );
    }
    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：创建GrpcClientChannel对象的方法，如果客户端访问的是proxy的netty服务端，那么就会调用RemotingChannelManager管理器的createChannel方法创建RemotingChannel
     */
    public GrpcClientChannel createChannel(ProxyContext ctx, String clientId) {
        //先根据clientId获取对应的GrpcClientChannel，如果不存在的话，就创建一个GrpcClientChannel对象放到clientIdChannelMap成员变量中
        return this.clientIdChannelMap.computeIfAbsent(clientId,
                k -> new GrpcClientChannel(proxyRelayService, grpcClientSettingsManager, this, ctx, clientId));
    }
    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：根据客户端Id得到对应的GrpcClientChannel对象的方法
     */
    public GrpcClientChannel getChannel(String clientId) {
        return clientIdChannelMap.get(clientId);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：从GrpcClientChannel缓存表中移除指定客户端的GrpcClientChannel对象的方法
     */
    public GrpcClientChannel removeChannel(String clientId) {
        return this.clientIdChannelMap.remove(clientId);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：当proxy服务端使用GrpcClientChannel对象向客户端发送请求后，会使用当前方法把为该请求创建的responseFuture对象存放到resultNonceFutureMap表中
     */
    public <T> String addResponseFuture(CompletableFuture<ProxyRelayResult<T>> responseFuture) {
        //为responseFuture对象生成一个唯一标志
        String nonce = this.nextNonce();
        //把responseFuture对象封装成ResultFuture对象放到resultNonceFutureMap表中
        this.resultNonceFutureMap.put(nonce, new ResultFuture<>(responseFuture));
        return nonce;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：从resultNonceFutureMap表中移除指定nonce对应的responseFuture对象的方法
     */
    public <T> CompletableFuture<ProxyRelayResult<T>> getAndRemoveResponseFuture(String nonce) {
        ResultFuture<T> resultFuture = this.resultNonceFutureMap.remove(nonce);
        if (resultFuture != null) {
            return resultFuture.future;
        }
        return null;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：为responseFuture对象生成唯一标识的方法
     */
    protected String nextNonce() {
        return String.valueOf(this.nonceIdGenerator.getAndIncrement());
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：扫描resultNonceFutureMap表并清除响应超时的responseFuture对象的方法
     */
    protected void scanExpireResultFuture() {
        //得到proxy的配置信息对象
        ProxyConfig proxyConfig = ConfigurationManager.getProxyConfig();
        //得到响应超时时间
        long timeOutMs = TimeUnit.SECONDS.toMillis(proxyConfig.getGrpcProxyRelayRequestTimeoutInSeconds());
        //接下来就可以遍历resultNonceFutureMap表了
        Set<String> nonceSet = this.resultNonceFutureMap.keySet();
        for (String nonce : nonceSet) {
            //根据nonce从resultNonceFutureMap表中获取对应的ResultFuture对象
            ResultFuture<?> resultFuture = this.resultNonceFutureMap.get(nonce);
            if (resultFuture == null) {
                continue;
            }
            //判断响应是否超时
            if (System.currentTimeMillis() - resultFuture.createTime > timeOutMs) {
                //超时则从resultNonceFutureMap表中移除这个ResultFuture对象
                resultFuture = this.resultNonceFutureMap.remove(nonce);
                if (resultFuture != null) {
                    //移除成功之后设置future响应超时状态
                    resultFuture.future.complete(new ProxyRelayResult<>(ResponseCode.SYSTEM_BUSY, "call remote timeout", null));
                }
            }
        }
    }
    @Override
    public void shutdown() throws Exception {
        this.scheduledExecutorService.shutdown();
    }

    @Override
    public void start() throws Exception {

    }

    protected static class ResultFuture<T> {
        public CompletableFuture<ProxyRelayResult<T>> future;
        public long createTime = System.currentTimeMillis();

        public ResultFuture(CompletableFuture<ProxyRelayResult<T>> future) {
            this.future = future;
        }
    }
}
